package org.csdn.titan.pre

import java.util.Properties

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.csdn.titan.beans.AppLogBean
import org.csdn.titan.utils.SparkUtils

/*
 * @Date:2021/2/22-19:34
 * @Author:fcc
 * @NickName:江南云朵
 * @QQ:46210953
 * @Description:预处理埋点日志
 */
object AppLogPreProcess {

  def main(args: Array[String]): Unit = {
    val session = SparkUtils.getSparkSession(this.getClass.getName)


    val logdata: Dataset[String] = session.read.textFile("E:\\restudy\\kaikeba\\learnspark\\scalaspark\\data\\yiee_logs\\2020-01-12\\app")
    //加载地理字典位位置信息，并广播出去
    val geodf:DataFrame = loadGeoInfo(session)
    val geoMap:scala.collection.Map[String,(String,String,String)] = geodf.rdd.map(row => {
      val geocode = row.getAs[String]("geocode")
      val provicename = row.getAs[String]("provicename")
      val cityname = row.getAs[String]("cityname")
      val districtname = row.getAs[String]("districtname")
      (geocode, (provicename, cityname, districtname))

    }).collectAsMap()
    val bc_geoMap = session.sparkContext.broadcast(geoMap)

    //加载guid数据并广播出去
    val guid_df = session.read.parquet("data/yiee_logs/2020-01-12/guidmap")
    val longToLong:scala.collection.Map[Long,Long] = guid_df.rdd.map(row => {
      val idcode = row.getAs[Long]("id_hashcode")
      val guid = row.getAs[Long]("guid")
      (idcode, guid)
    }).collectAsMap()
    val bc_guid = session.sparkContext.broadcast(longToLong)

    import session.implicits._
    import scala.collection.JavaConverters._
    val df =logdata.map(line => {

      try {
        val jsonObj = JSON.parseObject(line)

        val user = jsonObj.getJSONObject("user")
        val phone = user.getJSONObject("phone")
        val loc = user.getJSONObject("loc")

        var guid = Long.MaxValue
        val eventid = jsonObj.getString("eventid")
        val event: Map[String, String] = jsonObj.getJSONObject("event").getInnerMap.asInstanceOf[java.util.Map[String,String]].asScala.toMap
        val uid = user.getString("uid")
        val imei = phone.getString("imei")
        val mac = phone.getString("mac")
        val imsi = phone.getString("imsi")
        val osName = phone.getString("osName")
        val osVer = phone.getString("osVer")
        val androidId = phone.getString("androidId")
        val resolution = phone.getString("resolution")
        val deviceType = phone.getString("deviceType")
        val deviceId = phone.getString("deviceId")
        val uuid = phone.getString("uuid")
        val areacode = loc.getString("areacode")
        val longtitude = loc.getDouble("longtitude")
        val latitude = loc.getDouble("latitude")
        val carrier = loc.getString("carrier")
        val netType = loc.getString("netType")
        val cid_sn = loc.getString("cid_sn")
        val sessionId = user.getString("sessionId")
        val timestamp = jsonObj.getLong("timestamp")

        //过滤掉日志中：uid|imei|uuid|mac|androidId|
        var sb = new StringBuilder("")
        sb.append(uid).append(imei).append(mac).append(androidId)

        var flagField = sb.toString()
        flagField = flagField.replaceAll("null","")

        //过滤掉日志中缺小关键字段：event|eventid|sessionId，缺任务一个都不行 的记录

        var bean:AppLogBean = null
        if(event != null && StringUtils.isNoneBlank(eventid) && StringUtils.isNoneBlank(sessionId) && StringUtils.isNoneBlank(flagField)) {
          bean= AppLogBean(
            guid,
            eventid,
            event,
            uid,
            imei,
            mac,
            imsi,
            osName,
            osVer,
            androidId,
            resolution,
            deviceType,
            deviceId,
            uuid,
            areacode,
            longtitude,
            latitude,
            carrier,
            netType,
            cid_sn,
            sessionId,
            timestamp
          )
        }
        bean
      } catch {
        case e: Exception => null
      }
    })
      .filter(_ != null)
      .map(bean => {
        //把省、市、区及guid集成进来
        val geoMap = bc_geoMap.value
        val guidMap = bc_guid.value
        val longtitude = bean.longtitude
        val latitude = bean.latitude
        val geocode = GeoHash.geoHashStringWithCharacterPrecision(latitude, longtitude, 5)
        val maybeTuple:Option[(String,String,String)] = geoMap.get(geocode)
        if(maybeTuple.isDefined) {
          val pcd = maybeTuple.get
          bean.province = pcd._1
          bean.city = pcd._2
          bean.district = pcd._3
        }
        val ids = Array(bean.uid, bean.imei, bean.uuid, bean.mac, bean.androidId)
        var find = false
        for(id <- ids if !find) {
          val maybeLong = guidMap.get(id.hashCode.toLong)
          if(maybeLong.isDefined) {
            bean.guid = maybeLong.get
            find = true
          }
        }
        bean

      })
      .filter(bean => bean.guid != Long.MaxValue)
      .toDF().where("province!='未知'")
    //df.show(50,false)
      .write
      .parquet("data/apppredlog/2020-01-12/beans")
  }

  def loadGeoInfo(session:SparkSession):DataFrame = {
    val url = "jdbc:mysql://localhost:3306/csdn_titan?characterEncoding=UTF-8";
    val table = "district_dicts"
    var pro = new Properties();
    import scala.collection.JavaConverters._
    val map = Map("user"->"root","password"-> "123456").asJava

    pro.putAll(map);
    val df = session.read.jdbc(url,table,pro);
    df
  }

}
